package org.infinispan.security.impl;

import java.lang.annotation.Annotation;
import java.util.Collection;
import java.util.EnumSet;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;

import javax.security.auth.Subject;
import javax.transaction.xa.XAResource;

import jakarta.transaction.TransactionManager;

import org.infinispan.AdvancedCache;
import org.infinispan.CacheCollection;
import org.infinispan.CachePublisher;
import org.infinispan.CacheSet;
import org.infinispan.LockedStream;
import org.infinispan.batch.BatchContainer;
import org.infinispan.cache.impl.AbstractDelegatingAdvancedCache;
import org.infinispan.cache.impl.InternalCache;
import org.infinispan.commons.api.query.ContinuousQuery;
import org.infinispan.commons.api.query.Query;
import org.infinispan.commons.dataconversion.MediaType;
import org.infinispan.configuration.cache.Configuration;
import org.infinispan.configuration.global.GlobalSecurityConfiguration;
import org.infinispan.container.DataContainer;
import org.infinispan.container.entries.CacheEntry;
import org.infinispan.context.Flag;
import org.infinispan.distribution.DistributionManager;
import org.infinispan.encoding.DataConversion;
import org.infinispan.expiration.ExpirationManager;
import org.infinispan.factories.ComponentRegistry;
import org.infinispan.lifecycle.ComponentStatus;
import org.infinispan.manager.EmbeddedCacheManager;
import org.infinispan.metadata.Metadata;
import org.infinispan.notifications.cachelistener.filter.CacheEventConverter;
import org.infinispan.notifications.cachelistener.filter.CacheEventFilter;
import org.infinispan.partitionhandling.AvailabilityMode;
import org.infinispan.remoting.rpc.RpcManager;
import org.infinispan.security.AuthorizationManager;
import org.infinispan.security.AuthorizationPermission;
import org.infinispan.security.PrincipalRoleMapper;
import org.infinispan.security.SecureCache;
import org.infinispan.security.actions.SecurityActions;
import org.infinispan.security.mappers.ClusterRoleMapper;
import org.infinispan.stats.Stats;
import org.infinispan.util.concurrent.locks.LockManager;

/**
 * SecureCacheImpl.
 *
 * @author Tristan Tarrant
 * @since 7.0
 */
public final class SecureCacheImpl<K, V> extends AbstractDelegatingAdvancedCache<K, V> implements SecureCache<K, V>, InternalCache<K, V> {

   private final AuthorizationManager authzManager;
   private final AdvancedCache<K, V> delegate;
   private final Subject subject;
   private final AuthorizationPermission writePermission;

   public SecureCacheImpl(AdvancedCache<K, V> delegate) {
      this(delegate, delegate.getAuthorizationManager(), null);
   }

   private SecureCacheImpl(AdvancedCache<K, V> delegate, AuthorizationManager authzManager, Subject subject) {
      super(delegate);
      this.authzManager = authzManager;
      this.delegate = delegate;
      this.subject = subject;
      this.writePermission = authzManager.getWritePermission();
   }

   @Override
   public AdvancedCache<K, V> getDelegate() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate;
   }

   @Override
   public String getCacheAvailability() {
      authzManager.checkPermission(subject, AuthorizationPermission.MONITOR);
      return super.getCacheAvailability();
   }

   @Override
   public void setCacheAvailability(String availabilityString) {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      super.setCacheAvailability(availabilityString);
   }

   @Override
   public Properties getConfigurationAsProperties() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return super.getConfigurationAsProperties();
   }

   @Override
   public boolean isRebalancingEnabled() {
      authzManager.checkPermission(subject, AuthorizationPermission.MONITOR);
      return super.isRebalancingEnabled();
   }

   @Override
   public void setRebalancingEnabled(boolean enabled) {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      super.setRebalancingEnabled(enabled);
   }

   @Override
   public AdvancedCache<K, V> withSubject(Subject subject) {
      if (this.subject == null) {
         return new SecureCacheImpl<>(delegate, authzManager, subject);
      } else {
         throw new IllegalArgumentException("Cannot set a Subject on a SecureCache more than once");
      }
   }

   @Override
   public boolean startBatch() {
      authzManager.checkPermission(subject, writePermission);
      return delegate.startBatch();
   }

   @Override
   public void addListener(Object listener) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      delegate.addListener(listener);
   }

   @Override
   public <C> void addListener(Object listener, CacheEventFilter<? super K, ? super V> filter, CacheEventConverter<? super K, ? super V, C> converter) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      delegate.addListener(listener, filter, converter);
   }

   @Override
   public <C> CompletionStage<Void> addListenerAsync(Object listener, CacheEventFilter<? super K, ? super V> filter,
         CacheEventConverter<? super K, ? super V, C> converter) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      return delegate.addListenerAsync(listener, filter, converter);
   }

   @Override
   public CompletionStage<Void> addListenerAsync(Object listener) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      return delegate.addListenerAsync(listener);
   }

   @Override
   public <C> void addFilteredListener(Object listener, CacheEventFilter<? super K, ? super V> filter, CacheEventConverter<? super K, ? super V, C> converter, Set<Class<? extends Annotation>> filterAnnotations) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      delegate.addFilteredListener(listener, filter, converter, filterAnnotations);
   }

   @Override
   public <C> CompletionStage<Void> addFilteredListenerAsync(Object listener,
         CacheEventFilter<? super K, ? super V> filter, CacheEventConverter<? super K, ? super V, C> converter,
         Set<Class<? extends Annotation>> filterAnnotations) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      return delegate.addFilteredListenerAsync(listener, filter, converter, filterAnnotations);
   }

   @Override
   public <C> void addStorageFormatFilteredListener(Object listener, CacheEventFilter<? super K, ? super V> filter, CacheEventConverter<? super K, ? super V, C> converter, Set<Class<? extends Annotation>> filterAnnotations) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      delegate.addStorageFormatFilteredListener(listener, filter, converter, filterAnnotations);
   }

   @Override
   public <C> CompletionStage<Void> addStorageFormatFilteredListenerAsync(Object listener, CacheEventFilter<? super K, ? super V> filter, CacheEventConverter<? super K, ? super V, C> converter, Set<Class<? extends Annotation>> filterAnnotations) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      return delegate.addStorageFormatFilteredListenerAsync(listener, filter, converter, filterAnnotations);
   }

   @Override
   public void shutdown() {
      authzManager.checkPermission(subject, AuthorizationPermission.LIFECYCLE);
      delegate.shutdown();
   }

   @Override
   public void start() {
      authzManager.checkPermission(subject, AuthorizationPermission.LIFECYCLE);
      delegate.start();
      String name = delegate.getName();

      EmbeddedCacheManager ecm = delegate.getCacheManager();
      GlobalSecurityConfiguration gsc = SecurityActions.getCacheManagerConfiguration(ecm).security();
      PrincipalRoleMapper prm = gsc.authorization().principalRoleMapper();
      ClusterRoleMapper.includeMapperDependency(prm, name);
   }

   @Override
   public void stop() {
      authzManager.checkPermission(subject, AuthorizationPermission.LIFECYCLE);
      delegate.stop();
   }

   @Override
   public CompletableFuture<V> putAsync(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAsync(key, value);
   }

   @Override
   public void endBatch(boolean successful) {
      authzManager.checkPermission(subject, writePermission);
      delegate.endBatch(successful);
   }

   @Override
   public void removeListener(Object listener) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      delegate.removeListener(listener);
   }

   @Override
   public CompletionStage<Void> removeListenerAsync(Object listener) {
      authzManager.checkPermission(subject, AuthorizationPermission.LISTEN);
      return delegate.removeListenerAsync(listener);
   }

   @Override
   public CompletableFuture<V> putAsync(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAsync(key, value, lifespan, unit);
   }

   @Override
   public AdvancedCache<K, V> withFlags(Flag... flags) {
      return new SecureCacheImpl<>(delegate.withFlags(flags), authzManager, subject);
   }

   @Override
   public AdvancedCache<K, V> withFlags(Collection<Flag> flags) {
      return new SecureCacheImpl<>(delegate.withFlags(flags), authzManager, subject);
   }

   @Override
   public AdvancedCache<K, V> noFlags() {
      return new SecureCacheImpl<>(delegate.noFlags(), authzManager, subject);
   }

   @Override
   public AdvancedCache<K, V> transform(Function<AdvancedCache<K, V>, ? extends AdvancedCache<K, V>> transformation) {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      AdvancedCache<K, V> newDelegate = delegate.transform(transformation);
      AdvancedCache<K, V> newInstance = newDelegate != delegate ? new SecureCacheImpl<>(newDelegate, authzManager, subject) : this;
      return transformation.apply(newInstance);
   }

   @Override
   public V putIfAbsent(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsent(key, value);
   }

   @Override
   public CompletableFuture<V> putAsync(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdle,
                                        TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAsync(key, value, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public String getName() {
      return delegate.getName();
   }

   @Override
   public String getVersion() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getVersion();
   }

   @Override
   public V put(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.put(key, value);
   }

   @Override
   public CompletableFuture<Void> putAllAsync(Map<? extends K, ? extends V> data) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAllAsync(data);
   }

   @Override
   public V put(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.put(key, value, lifespan, unit);
   }

   @Override
   public boolean remove(Object key, Object value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.remove(key, value);
   }

   @Override
   public CompletableFuture<Void> putAllAsync(Map<? extends K, ? extends V> data, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAllAsync(data, lifespan, unit);
   }

   @Override
   public V putIfAbsent(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsent(key, value, lifespan, unit);
   }

   @Override
   public CompletableFuture<Void> putAllAsync(Map<? extends K, ? extends V> data, long lifespan, TimeUnit lifespanUnit,
                                              long maxIdle, TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAllAsync(data, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public void putAll(Map<? extends K, ? extends V> map, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putAll(map, lifespan, unit);
   }

   @Override
   public boolean replace(K key, V oldValue, V newValue) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, oldValue, newValue);
   }

   @Override
   public CompletableFuture<Void> clearAsync() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_WRITE);
      return delegate.clearAsync();
   }

   @Override
   public V replace(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, value, lifespan, unit);
   }

   @Override
   public CompletableFuture<V> putIfAbsentAsync(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsentAsync(key, value);
   }

   @Override
   public boolean replace(K key, V oldValue, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, oldValue, value, lifespan, unit);
   }

   @Override
   public CompletableFuture<V> putIfAbsentAsync(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsentAsync(key, value, lifespan, unit);
   }

   @Override
   public V replace(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, value);
   }

   @Override
   public ExpirationManager<K, V> getExpirationManager() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getExpirationManager();
   }

   @Override
   public V put(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.put(key, value, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public void putForExternalRead(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putForExternalRead(key, value);
   }

   @Override
   public void putForExternalRead(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putForExternalRead(key, value);
   }

   @Override
   public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.compute(key, remappingFunction);
   }

   @Override
   public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.merge(key, value, remappingFunction);
   }

   @Override
   public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.merge(key, value, remappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.merge(key, value, remappingFunction, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public V merge(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.merge(key, value, remappingFunction, metadata);
   }

   @Override
   public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.compute(key, remappingFunction, metadata);
   }

   @Override
   public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.compute(key, remappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public V compute(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.compute(key, remappingFunction, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresent(key, remappingFunction);
   }

   @Override
   public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresent(key, remappingFunction, metadata);
   }

   @Override
   public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresent(key, remappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public void forEach(BiConsumer<? super K, ? super V> action) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      super.forEach(action);
   }

   @Override
   public V computeIfPresent(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresent(key, remappingFunction, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsent(key, mappingFunction);
   }

   @Override
   public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsent(key, mappingFunction, metadata);
   }

   @Override
   public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsent(key, mappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public V computeIfAbsent(K key, Function<? super K, ? extends V> mappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsent(key, mappingFunction, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public CompletableFuture<V> computeAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeAsync(key, remappingFunction, metadata);
   }

   @Override
   public CompletableFuture<V> computeIfPresentAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresentAsync(key, remappingFunction, metadata);
   }

   @Override
   public CompletableFuture<V> computeIfAbsentAsync(K key, Function<? super K, ? extends V> mappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsentAsync(key, mappingFunction, metadata);
   }

   @Override
   public CompletableFuture<V> mergeAsync(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.mergeAsync(key, value, remappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public CompletableFuture<V> mergeAsync(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.mergeAsync(key, value, remappingFunction, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public CompletableFuture<V> mergeAsync(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.mergeAsync(key, value, remappingFunction, metadata);
   }

   @Override
   public CompletableFuture<V> computeAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeAsync(key, remappingFunction);
   }

   @Override
   public CompletableFuture<V> computeAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeAsync(key, remappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public CompletableFuture<V> computeAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdle, TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeAsync(key, remappingFunction, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public CompletableFuture<V> computeIfAbsentAsync(K key, Function<? super K, ? extends V> mappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsentAsync(key, mappingFunction);
   }

   @Override
   public CompletableFuture<V> computeIfAbsentAsync(K key, Function<? super K, ? extends V> mappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsentAsync(key, mappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public CompletableFuture<V> computeIfAbsentAsync(K key, Function<? super K, ? extends V> mappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdle, TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfAbsentAsync(key, mappingFunction, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public CompletableFuture<V> computeIfPresentAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresentAsync(key, remappingFunction);
   }

   @Override
   public CompletableFuture<V> computeIfPresentAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresentAsync(key, remappingFunction, lifespan, lifespanUnit);
   }

   @Override
   public CompletableFuture<V> computeIfPresentAsync(K key, BiFunction<? super K, ? super V, ? extends V> remappingFunction, long lifespan, TimeUnit lifespanUnit, long maxIdle, TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.computeIfPresentAsync(key, remappingFunction, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public CompletableFuture<V> mergeAsync(K key, V value, BiFunction<? super V, ? super V, ? extends V> remappingFunction) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.mergeAsync(key, value, remappingFunction);
   }

   @Override
   public void putForExternalRead(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putForExternalRead(key, value);
   }

   @Override
   public void putForExternalRead(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdle, TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putForExternalRead(key, value);
   }

   @Override
   public ComponentRegistry getComponentRegistry() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return ComponentRegistry.of(delegate);
   }

   @Override
   public DistributionManager getDistributionManager() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getDistributionManager();
   }

   @Override
   public AuthorizationManager getAuthorizationManager() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return new AuthorizationManager() {
         @Override
         public void checkPermission(AuthorizationPermission permission) {
            authzManager.checkPermission(subject, permission);
         }

         @Override
         public void checkPermission(Subject subject, AuthorizationPermission permission) {
            authzManager.checkPermission(subject, permission);
         }

         @Override
         public void checkPermission(AuthorizationPermission permission, String role) {
            authzManager.checkPermission(subject, permission, role);
         }

         @Override
         public void checkPermission(Subject subject, AuthorizationPermission permission, String role) {
            authzManager.checkPermission(subject, permission, role);
         }

         @Override
         public EnumSet<AuthorizationPermission> getPermissions(Subject subject) {
            return authzManager.getPermissions(subject);
         }

         @Override
         public AuthorizationPermission getWritePermission() {
            return authzManager.getWritePermission();
         }

         @Override
         public void doIf(Subject subject, AuthorizationPermission permission, Runnable runnable) {
            authzManager.doIf(subject, permission, runnable);
         }

         @Override
         public boolean isPermissive() {
            return authzManager.isPermissive();
         }
      };
   }

   @Override
   public AdvancedCache<K, V> lockAs(Object lockOwner) {
      return new SecureCacheImpl<>(delegate.lockAs(lockOwner));
   }

   @Override
   public CompletableFuture<V> putIfAbsentAsync(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdle,
                                                TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsentAsync(key, value, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public boolean isEmpty() {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.isEmpty();
   }

   @Override
   public boolean lock(K... keys) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.lock(keys);
   }

   @Override
   public boolean containsKey(Object key) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.containsKey(key);
   }

   @Override
   public CompletableFuture<Boolean> containsKeyAsync(K key) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return super.containsKeyAsync(key);
   }

   @Override
   public V putIfAbsent(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsent(key, value, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public boolean lock(Collection<? extends K> keys) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.lock(keys);
   }

   @Override
   public CompletableFuture<V> removeAsync(Object key) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.removeAsync(key);
   }

   @Override
   public CompletableFuture<CacheEntry<K, V>> removeAsyncEntry(Object key) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.removeAsyncEntry(key);
   }

   @Override
   public boolean containsValue(Object value) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.containsValue(value);
   }

   @Override
   public void putAll(Map<? extends K, ? extends V> map, long lifespan, TimeUnit lifespanUnit, long maxIdleTime,
                      TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putAll(map, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public CompletableFuture<Boolean> removeAsync(Object key, Object value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.removeAsync(key, value);
   }

   @Override
   public void evict(K key) {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      delegate.evict(key);
   }

   @Override
   public CompletableFuture<V> replaceAsync(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, value);
   }

   @Override
   public void replaceAll(BiFunction<? super K, ? super V, ? extends V> function) {
      authzManager.checkPermission(subject, writePermission);
      super.replaceAll(function);
   }

   @Override
   public RpcManager getRpcManager() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getRpcManager();
   }

   @Override
   public V replace(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdleTime, TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, value, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public V get(Object key) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.get(key);
   }

   @Override
   public CompletableFuture<V> replaceAsync(K key, V value, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, value, lifespan, unit);
   }

   @Override
   public BatchContainer getBatchContainer() {
      authzManager.checkPermission(subject, writePermission);
      return delegate.getBatchContainer();
   }

   @Override
   public Configuration getCacheConfiguration() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getCacheConfiguration();
   }

   @Override
   public EmbeddedCacheManager getCacheManager() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getCacheManager();
   }

   @Override
   public AdvancedCache<K, V> getAdvancedCache() {
      return this;
   }

   @Override
   public ComponentStatus getStatus() {
      return delegate.getStatus();
   }

   @Override
   public AvailabilityMode getAvailability() {
      return delegate.getAvailability();
   }

   @Override
   public void setAvailability(AvailabilityMode availabilityMode) {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      delegate.setAvailability(availabilityMode);
   }

   @Override
   public CacheSet<CacheEntry<K, V>> cacheEntrySet() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.cacheEntrySet();
   }

   @Override
   public LockedStream<K, V> lockedStream() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_WRITE);
      return delegate.lockedStream();
   }

   @Override
   public CompletableFuture<Boolean> removeLifespanExpired(K key, V value, Long lifespan) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.removeLifespanExpired(key, value, lifespan);
   }

   @Override
   public CompletableFuture<Boolean> removeMaxIdleExpired(K key, V value) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.removeMaxIdleExpired(key, value);
   }

   @Override
   public <K1, V1> AdvancedCache<K1, V1> withMediaType(MediaType keyMediaType, MediaType valueMediaType) {
      return new SecureCacheImpl<>(delegate.withMediaType(keyMediaType, valueMediaType), authzManager, subject);
   }

   @Override
   public AdvancedCache<K, V> withStorageMediaType() {
      return new SecureCacheImpl<>(delegate.withStorageMediaType(), authzManager, subject);
   }

   @Override
   public AdvancedCache rewrap(AdvancedCache newDelegate) {
      return new SecureCacheImpl(newDelegate, authzManager, subject);
   }

   @Override
   public DataConversion getKeyDataConversion() {
      return delegate.getKeyDataConversion();
   }

   @Override
   public DataConversion getValueDataConversion() {
      return delegate.getValueDataConversion();
   }

   @Override
   public int size() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.size();
   }

   @Override
   public CompletableFuture<Long> sizeAsync() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.sizeAsync();
   }

   @Override
   public boolean replace(K key, V oldValue, V value, long lifespan, TimeUnit lifespanUnit, long maxIdleTime,
                          TimeUnit maxIdleTimeUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, oldValue, value, lifespan, lifespanUnit, maxIdleTime, maxIdleTimeUnit);
   }

   @Override
   public DataContainer<K, V> getDataContainer() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getDataContainer();
   }

   @Override
   public CompletableFuture<V> replaceAsync(K key, V value, long lifespan, TimeUnit lifespanUnit, long maxIdle,
                                            TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, value, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public TransactionManager getTransactionManager() {
      return delegate.getTransactionManager();
   }

   @Override
   public CacheSet<K> keySet() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.keySet();
   }

   @Override
   public V remove(Object key) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.remove(key);
   }

   @Override
   public <T> Query<T> query(String query) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.query(query);
   }

   @Override
   public ContinuousQuery<K, V> continuousQuery() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.continuousQuery();
   }

   @Override
   public Map<K, V> getAll(Set<?> keys) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.getAll(keys);
   }

   @Override
   public CompletableFuture<Map<K, V>> getAllAsync(Set<?> keys) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.getAllAsync(keys);
   }

   @Override
   public LockManager getLockManager() {
      authzManager.checkPermission(subject, writePermission);
      return delegate.getLockManager();
   }

   @Override
   public CompletableFuture<Boolean> replaceAsync(K key, V oldValue, V newValue) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, oldValue, newValue);
   }

   @Override
   public Stats getStats() {
      authzManager.checkPermission(subject, AuthorizationPermission.MONITOR);
      return delegate.getStats();
   }

   @Override
   public XAResource getXAResource() {
      authzManager.checkPermission(subject, AuthorizationPermission.ADMIN);
      return delegate.getXAResource();
   }

   @Override
   public ClassLoader getClassLoader() {
      return delegate.getClassLoader();
   }

   @Override
   public CompletableFuture<Boolean> replaceAsync(K key, V oldValue, V newValue, long lifespan, TimeUnit unit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, oldValue, newValue, lifespan, unit);
   }

   @Override
   public CacheCollection<V> values() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.values();
   }

   @Override
   public CompletableFuture<Boolean> replaceAsync(K key, V oldValue, V newValue, long lifespan, TimeUnit lifespanUnit,
                                                  long maxIdle, TimeUnit maxIdleUnit) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, oldValue, newValue, lifespan, lifespanUnit, maxIdle, maxIdleUnit);
   }

   @Override
   public CacheSet<Entry<K, V>> entrySet() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.entrySet();
   }

   @Override
   public CompletableFuture<V> getAsync(K key) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.getAsync(key);
   }

   @Override
   public V getOrDefault(Object key, V defaultValue) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return super.getOrDefault(key, defaultValue);
   }

   @Override
   public V put(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.put(key, value, metadata);
   }

   @Override
   public void putAll(Map<? extends K, ? extends V> m, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putAll(m, metadata);
   }

   @Override
   public CompletableFuture<Void> putAllAsync(Map<? extends K, ? extends V> map, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAllAsync(map, metadata);
   }

   @Override
   public void putAll(Map<? extends K, ? extends V> m) {
      authzManager.checkPermission(subject, writePermission);
      delegate.putAll(m);
   }

   @Override
   public V replace(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, value, metadata);
   }

   @Override
   public CompletableFuture<V> replaceAsync(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, value, metadata);
   }

   @Override
   public CompletableFuture<CacheEntry<K, V>> replaceAsyncEntry(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsyncEntry(key, value, metadata);
   }

   @Override
   public boolean replace(K key, V oldValue, V newValue, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replace(key, oldValue, newValue, metadata);
   }

   @Override
   public CompletableFuture<Boolean> replaceAsync(K key, V oldValue, V newValue, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.replaceAsync(key, oldValue, newValue, metadata);
   }

   @Override
   public void clear() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_WRITE);
      delegate.clear();
   }

   @Override
   public V putIfAbsent(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsent(key, value, metadata);
   }

   @Override
   public CompletableFuture<V> putIfAbsentAsync(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsentAsync(key, value, metadata);
   }

   @Override
   public CompletableFuture<CacheEntry<K, V>> putIfAbsentAsyncEntry(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putIfAbsentAsyncEntry(key, value, metadata);
   }

   @Override
   public CompletableFuture<V> putAsync(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAsync(key, value, metadata);
   }

   @Override
   public CompletableFuture<CacheEntry<K, V>> putAsyncEntry(K key, V value, Metadata metadata) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.putAsyncEntry(key, value, metadata);
   }
   @Override
   public CacheEntry<K, V> getCacheEntry(Object key) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.getCacheEntry(key);
   }

   @Override
   public CompletableFuture<CacheEntry<K, V>> getCacheEntryAsync(Object key) {
      authzManager.checkPermission(subject, AuthorizationPermission.READ);
      return delegate.getCacheEntryAsync(key);
   }

   @Override
   public Map<K, CacheEntry<K, V>> getAllCacheEntries(Set<?> keys) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.getAllCacheEntries(keys);
   }

   @Override
   public Map<K, V> getAndPutAll(Map<? extends K, ? extends V> map) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_WRITE);
      return delegate.getAndPutAll(map);
   }

   @Override
   public Map<K, V> getGroup(String groupName) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.getGroup(groupName);
   }

   @Override
   public void removeGroup(String groupName) {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_WRITE);
      delegate.removeGroup(groupName);
   }

   @Override
   public boolean equals(Object o) {
      return delegate.equals(o);
   }

   @Override
   public int hashCode() {
      return delegate.hashCode();
   }

   @Override
   public String toString() {
      return "Secure " + delegate;
   }

   @Override
   public CompletionStage<Boolean> touch(Object key, int segment, boolean touchEvenIfExpired) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.touch(key, segment, touchEvenIfExpired);
   }

   @Override
   public CompletionStage<Boolean> touch(Object key, boolean touchEvenIfExpired) {
      authzManager.checkPermission(subject, writePermission);
      return delegate.touch(key, touchEvenIfExpired);
   }

   @Override
   public CachePublisher<K, V> cachePublisher() {
      authzManager.checkPermission(subject, AuthorizationPermission.BULK_READ);
      return delegate.cachePublisher();
   }
}
